package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.TaskInfo;
import com.atguigu.userprofile.bean.TaskTagRule;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.dao.TaskInfoDAO;
import com.atguigu.userprofile.dao.TaskTagRuleDAO;
import com.atguigu.userprofile.util.MyPropertiesUtil;
import com.atguigu.userprofile.util.MySQLUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

public class TaskSQLApp {


    //1  读取前置: 获得本标签的命名、编码、SQL、映射…信息
    //
    // 2  准备标签表
    //
    // 3   组合一条insert select 语句 来进行hive的读取写入
    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf().setAppName("task_sql_app");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        String taskId=args[0];
        String busiDate=args[1];

        //1  读取前置 :获得本标签的命名、编码、SQL、映射…信息
        //去mysql中查询 tag_info 、 task_info 、task_tag_rule
        // mysql查询工具
        //tag_info    标签表
        TagInfo tagInfo = TagInfoDAO.getTagInfoByTaskId(taskId);
        System.out.println(tagInfo);
        //task_info    任务表
        TaskInfo taskInfo = TaskInfoDAO.getTaskInfo(taskId);
        System.out.println(taskInfo);
        //task_tag_rule   查询结果和四级标签的映射
        List<TaskTagRule> taskTagRuleList = TaskTagRuleDAO.getTaskTagRuleList(taskId);
        System.out.println(taskTagRuleList);

        //2  准备标签表
        //   多个标签  每个标签一张表 ？  所有标签一张表？  都可以   所有标签建一张表 标签之间就需要用分区隔离


        //    create table if not exists  $tablename
        //    (uid string ,tag_value  $field_type)
        //    分区   日期  partitioned by (dt string)
        //     comment ''
        //     压缩 ？ 格式？   不压缩  普通文本  ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
        //    location  'hdfs://根目录/库目录/表目录

        //标签表名用 tag_code
        //字段类型  标签值的类型 tag_value_type
        // 备注 tag_name

        String tableName= tagInfo.getTagCode().toLowerCase();

        String fieldType="";

        if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_LONG)){
            fieldType="bigint";
        }else if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_DECIMAL)){
            fieldType="decimal";
        }else if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_STRING)){
            fieldType="string";
        }else if(tagInfo.getTagValueType().equals(ConstCodes.TAG_VALUE_TYPE_DATE)){
            fieldType="string";
        }
        String comment =tagInfo.getTagName();

        Properties properties = MyPropertiesUtil.load("config.properties");

        String hdfsPath = properties.getProperty("hdfs-store.path");
        String upName = properties.getProperty("user-profile.dbname");
        String dwName = properties.getProperty("data-warehouse.dbname");


        String createTableSQL="     create table if not exists   "+upName+"." +tableName+
                "            (uid string ,tag_value  "+fieldType+")\n" +
                "             partitioned by (dt string)\n" +
                "             comment '"+comment+"' " +
                "            ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'\n" +
                "             location  '"+hdfsPath+"/"+upName+"/"+tableName+"'";
        System.out.println(createTableSQL);
        sparkSession.sql(createTableSQL);


        // 3   组合一条insert select 语句 来进行hive的读取写入
        // select  uid ,  case  query_value when  'M' then '男' when 'F' then '女'  when 'U' then '未知' end tag_value
        // from (
        //   select  id as uid ,  nvl( gender,'U')  as    query_value  from dim_user_zip  where dt='9999-12-31'
        // ) tt


        // 作业 ： 兼容型修改  兼容没有四级标签的情况

        String tagValueSQL="";
        if(taskTagRuleList.size()>0){
            List<String> whenThenList = taskTagRuleList.stream().map(taskTagRule -> "when '" + taskTagRule.getQueryValue() + "' then '" + taskTagRule.getSubTagValue()+"'").collect(Collectors.toList());
            String whenThenSQL = StringUtils.join(whenThenList, " ");
            String caseSQL="case query_value  "+whenThenSQL+" end  as tag_value";
            tagValueSQL=caseSQL;
        }else{
            tagValueSQL= "query_value as tag_value ";
        }


        // 作业：替换业务日期
        String taskSql = taskInfo.getTaskSql().replace("$dt", busiDate);

        String selectSQL=" select  uid ,"+tagValueSQL+
                "  from ( "+taskSql+") tt";
       // System.out.println( selectSQL);
        // insert overwrite table $tableName partition( dt ='$busiDate')
        // $selectSQL



        String insertSQL = " insert overwrite table "+upName+"."+tableName+" partition( dt ='"+busiDate+"') "+selectSQL;

        System.out.println( insertSQL);
        sparkSession.sql("use "+dwName);

        sparkSession.sql(insertSQL);
    }
}
